package com.inno.subject.domain.websocket;


import com.inno.subject.domain.listener.SocketCloseListener;
import com.inno.subject.domain.utils.ApplicationContextUtil;
import com.inno.subject.domain.utils.ChatClientUtil;
import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicInteger;

@ServerEndpoint(value = "/socket/subject/{userId}",configurator = WebSocketConfigurator.class)
@Slf4j
@Component
public class SubjectSocket {
    /**
     * 在线人数
     */
    private static final AtomicInteger ONLINE_COUNT = new AtomicInteger(0);
    /**
     * 在线客户端
     */
    private static final ConcurrentHashMap<Long, Session> ONLINE_CLIENTS = new ConcurrentHashMap<>();

    /**
     * 离线消息队列
     */
    private static final ConcurrentHashMap<Long, ConcurrentLinkedQueue<String>> MESSAGE_QUEUE = new ConcurrentHashMap<>();


    public SubjectSocket() {
    }

    @OnOpen
    public void onOpen(Session session) {
        Long userId = getPathParam(session);
        log.info("用户{}连接", userId);
        if (!ONLINE_CLIENTS.containsKey(userId)) {
            ONLINE_CLIENTS.put(userId, session);
            ONLINE_COUNT.incrementAndGet();
            log.info("onOpen,当前在线人数:{}", ONLINE_COUNT.get());
        }

        if (MESSAGE_QUEUE.containsKey(userId)) {
            ConcurrentLinkedQueue<String> messages = MESSAGE_QUEUE.get(userId);
            for (String message : messages) {
                sendMessage(session, message);
            }
            MESSAGE_QUEUE.remove(userId);
        }

    }


    @OnClose
    public void onClose(Session session) {
        Long pathId = getPathParam(session);
        log.info("用户{}断开连接", pathId);

        if (ONLINE_CLIENTS.containsKey(pathId)) {
            ONLINE_CLIENTS.remove(pathId);
            int i = ONLINE_COUNT.decrementAndGet();
            log.info("onClose,当前在线人数:{}",  i);
        }
        // 清空历史消息
        new SocketCloseListener().invoke(pathId);
    }

    private Long getPathParam(Session session) {
        try {
            String userId = session.getPathParameters().get("userId");
            if (StringUtils.isBlank(userId)) {
                sendMessage(session, "未登录，不能连接");
            }
            Long pathId = Long.valueOf(userId);
            return pathId;
        } catch (Throwable e) {
            log.error("获取用户id失败，原因：{}", e.getMessage());
            sendMessage(session, "连接出错");
            throw new RuntimeException("连接出错");
        }
    }

    @OnMessage
    public void onMessage(String message, Session session) {
        // 心跳
        if ("ping".equals(message)) {
            sendMessage(session, "pong");
        }
    }

    @OnError
    public void onError(Session session, Throwable error) {
        log.error("Socket:{},发生错误,错误原因{}", session, error.getMessage(), error);
        try {
            session.close();
        } catch (Exception e) {
            log.error("onError.Exception{}", e.getMessage(), e);
        }
    }


    public synchronized void sendMessage(Session session, String message) {
        try {
            session.getBasicRemote().sendText(message);
        } catch (IOException e) {
            log.error("发送消息失败，原因：", e);
        }
    }

    public synchronized Session getSession(Long userId) {
        return ONLINE_CLIENTS.get(userId);
    }


    /**
     * 暂存离线消息
     */
    public void putOfflineMessage(Long userId, String message) {
        ConcurrentLinkedQueue<String> messages = MESSAGE_QUEUE.get(userId);
        if (messages == null) {
            messages = new ConcurrentLinkedQueue<>();
        }
        messages.add(message);
        MESSAGE_QUEUE.put(userId, messages);
    }

}
